use std::{
    sync::{
        atomic::{AtomicU8, Ordering},
        Arc,
    },
    time::Duration,
};

use cframe::{
    ctype::{self},
    make, make_empty, make_query_task,
};
use dashmap::DashMap;
use jni::objects::{GlobalRef, JObject, JValue};

#[cfg(unix)]
use libc::{flock, LOCK_EX, LOCK_NB};

use log::debug;
use reader::read_frame;
use tokio::{
    io::{split, AsyncReadExt, AsyncWriteExt, BufReader, ReadHalf, WriteHalf},
    sync::{mpsc, oneshot, Mutex, Semaphore},
    time::timeout,
};

use tokio_serial::{SerialPortBuilderExt, SerialStream};

#[cfg(unix)]
use std::os::fd::AsRawFd;

use crate::{
    error::{new_err, AppErr, IoErr}, utils::global::get_java_vm,
};

mod cframe;
mod reader;

pub use cframe::RecvFrame;

pub enum RespSender {
    Oneshot(oneshot::Sender<RecvFrame>),
    Mpsc(mpsc::Sender<RecvFrame>),
}

pub type RespMap = DashMap<u16, RespSender>;
pub type Reader = BufReader<ReadHalf<SerialStream>>;

const CB_NOTIFY: &'static str = "notify";
const CB_NOTIFY_ACK: &'static str = "notifyAck";

pub struct SerialPortConn {
    writer: Mutex<WriteHalf<SerialStream>>,
    resp_map: RespMap,
    seq: AtomicU8,
    exit_sem: Semaphore,
    cb: GlobalRef,
}

pub type SharedConn = Arc<SerialPortConn>;

impl Drop for SerialPortConn {
    fn drop(&mut self) {
        debug!("serial port drop");
    }
}

pub fn new_serial_port(name: &str, baud_rate: u32, cb: GlobalRef) -> Result<SharedConn, AppErr> {
    let stream = tokio_serial::new(name, baud_rate)
        .open_native_async()
        .map_err(|_| new_err("打开串口失败"))?;

    #[cfg(unix)]
    {
        let fd = stream.as_raw_fd();
        let ret = unsafe { flock(fd, LOCK_EX | LOCK_NB) };
        if ret != 0 {
            return Err(new_err("串口重复打开"));
        }
    }

    let (reader, writer) = split(stream);
    let reader = BufReader::new(reader);
    let conn = Arc::new(SerialPortConn {
        writer: Mutex::new(writer),
        resp_map: DashMap::new(),
        seq: AtomicU8::new(0),
        exit_sem: Semaphore::new(0),
        cb,
    });
    tokio::spawn(read_task(conn.clone(), reader));
    Ok(conn)
}

async fn read_loop<R: AsyncReadExt + Unpin>(
    conn: &SerialPortConn,
    r: &mut R,
) -> Result<(), AppErr> {
    loop {
        let frame = tokio::select! {
            frame = read_frame(r) => {
                frame
            }
            _ = conn.exit_sem.acquire() => {
                break
            }
        };
        let frame = match frame {
            Ok(f) => f,
            Err(e) => return Err(e),
        };

        match frame.ctype() {
            ctype::ACK | ctype::PONG | ctype::SIMPLE_RES => {
                conn.on_resp(frame);
            }
            ctype::PING => {
                conn.pong(frame.seq()).await?;
            }
            ctype::NOTIFY => {
                _ = conn.notify(CB_NOTIFY, frame);
            }
            ctype::NOTIFY_ACK => {
                conn.ack(frame.seq()).await?;
                _ = conn.notify(CB_NOTIFY_ACK, frame);
            }
            ctype::RES => {
                conn.ack(frame.seq()).await?;
                conn.on_resp(frame);
            }
            _ => {}
        };
    }

    Ok(())
}

async fn read_task(conn: SharedConn, mut reader: Reader) {
    let r = read_loop(&conn, &mut reader).await;
    if let Err(e) = r {
        debug!("read err:{}", e);
    }
    debug!("read exit");
}

impl SerialPortConn {
    pub fn close(&self) {
        self.exit_sem.add_permits(1);
    }

    pub async fn request(&self, cmd: u8, data: &[u8]) -> Result<RecvFrame, AppErr> {
        let seq = self.create_seq();
        self.session(seq).await?;
        let mut rx = self.create_resp_mpsc(ctype::RES, seq);
        self.req(seq, cmd, data).await?;

        loop {
            match timeout(Duration::from_secs(3), rx.recv()).await {
                Ok(Some(v)) => return Ok(v),
                _ => {}
            };
            match self.query_task(seq).await {
                Ok(_) => continue,
                _ => {}
            };
            match timeout(Duration::from_secs(3), rx.recv()).await {
                Ok(Some(v)) => return Ok(v),
                _ => return Err(new_err("resp timeout")),
            };
        }
    }

    async fn req(&self, seq: u8, cmd: u8, data: &[u8]) -> Result<(), AppErr> {
        let frame = make(ctype::REQ, seq, cmd, data);
        self.exec(&frame, ctype::ACK, seq).await?;
        Ok(())
    }

    async fn session(&self, seq: u8) -> Result<(), AppErr> {
        let frame = make_empty(ctype::SESSION, seq);
        self.exec(&frame, ctype::ACK, seq).await?;
        Ok(())
    }

    async fn query_task(&self, dst_seq: u8) -> Result<(), AppErr> {
        let seq = self.create_seq();
        let frame = make_query_task(seq, dst_seq);
        let ret = self.exec(&frame, ctype::SIMPLE_RES, seq).await?;
        if ret.ec() != 0 {
            return Err(new_err("task finish"));
        }
        Ok(())
    }

    pub async fn simple_req(&self, cmd: u8, data: &[u8]) -> Result<RecvFrame, AppErr> {
        let seq = self.create_seq();
        let frame = make(ctype::SIMPLE_REQ, seq, cmd, data);
        let ret = self.exec(&frame, ctype::SIMPLE_RES, seq).await?;
        Ok(ret)
    }

    pub async fn ping(&self) -> Result<(), AppErr> {
        let seq = self.create_seq();
        let frame = make_empty(ctype::PING, seq);
        self.exec(&frame, ctype::PONG, seq).await?;
        Ok(())
    }

    async fn exec(&self, frame: &[u8], r_ctype: u8, seq: u8) -> Result<RecvFrame, AppErr> {
        for _ in 0..3 {
            let rx = self.create_resp(r_ctype, seq);
            self.write(frame).await?;
            match timeout(Duration::from_millis(500), rx).await {
                Ok(Ok(ret)) => return Ok(ret),
                _ => {}
            };
        }
        Err(new_err("recv超时"))
    }

    async fn write(&self, buf: &[u8]) -> Result<(), IoErr> {
        let mut w = self.writer.lock().await;
        w.write_all(buf).await?;
        Ok(())
    }

    fn create_seq(&self) -> u8 {
        self.seq.fetch_add(1, Ordering::Relaxed)
    }

    fn create_resp(&self, ctype: u8, seq: u8) -> oneshot::Receiver<RecvFrame> {
        let id = ((ctype as u16) << 8) + (seq as u16);
        let (tx, rx) = oneshot::channel();
        self.resp_map.insert(id, RespSender::Oneshot(tx));
        rx
    }

    fn create_resp_mpsc(&self, ctype: u8, seq: u8) -> mpsc::Receiver<RecvFrame> {
        let id = ((ctype as u16) << 8) + (seq as u16);
        let (tx, rx) = mpsc::channel(1);
        self.resp_map.insert(id, RespSender::Mpsc(tx));
        rx
    }

    fn on_resp(&self, frame: RecvFrame) {
        let id = frame.id();
        let tx = self.resp_map.remove(&id);
        if let Some(tx) = tx {
            match tx.1 {
                RespSender::Oneshot(tx) => _ = tx.send(frame),
                RespSender::Mpsc(tx) => _ = tx.try_send(frame),
            };
        }
    }

    async fn ack(&self, seq: u8) -> Result<(), IoErr> {
        let frame = make_empty(ctype::ACK, seq);
        self.write(&frame).await?;
        Ok(())
    }

    async fn pong(&self, seq: u8) -> Result<(), IoErr> {
        let frame = make_empty(ctype::PONG, seq);
        self.write(&frame).await?;
        Ok(())
    }

    fn notify(&self, name: &str, frame: RecvFrame) -> Result<(), AppErr> {
        let mut env =  get_java_vm().attach_current_thread_permanently()?;
        let buf = frame.new_jbyte_array(&mut env);
        let raw = unsafe { JObject::from_raw(buf.as_raw()) };

        let args = [JValue::Object(&raw)];
        _ = env.call_method(&self.cb, name, "([B)V", &args);
        Ok(())
    }
}
